package com.shujia.mllib

import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg.{SparseVector, Vectors}
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo6ProImage {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .master("local[8]")
      .appName("person")
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()

    val image: DataFrame = spark
      .read
      .format("image")
      .load("D:\\课件\\机器学习数据\\手写数字\\131.jpg")

    import spark.implicits._

    val data: DataFrame = image
      .select($"image.origin", $"image.data")
      .as[(String, Array[Byte])]
      .map {
        case (name: String, data: Array[Byte]) => {
          val ints: Array[Int] = data.map(b => b.toInt)

          //将数据归一化
          val result: Array[Double] = ints.map(i => {
            if (i < 0) {
              1.0
            } else {
              0.0
            }
          })

          //将数组转换成向量
          val fea: SparseVector = Vectors.dense(result).toSparse

          val filename: String = name.split("/").last

          (filename, fea)
        }
      }.toDF("name", "features")


    /**
      * 加载模型
      *
      */

    val model: LogisticRegressionModel = LogisticRegressionModel.load("spark/data/image_model")

    //预测
    val frame: DataFrame = model.transform(data)

    frame.show()
  }

}
